package com.econ.powercloud.ems.util;

import com.econ.powercloud.common.base.facade.util.JacksonUtil;
import com.econ.powercloud.datahandle.facade.constants.KafkaHeaderConstant;
import com.econ.powercloud.ems.common.enums.CacheTopicType;
import com.econ.powercloud.ems.common.enums.OperationTypeEnum;
import com.econ.powercloud.ems.common.pojo.CacheOperationVO;
import com.google.common.collect.Lists;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.converter.MessagingMessageConverter;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;

/**
 * @author hcx
 * @since 2023/7/31 16:46
 **/
@Component
public class SendKafkaDateUtil {
    private static MessagingMessageConverter converter = new MessagingMessageConverter();

    @Autowired
    private KafkaTemplate template;

    @Value("${dataflow.producer.econ-common-bus-topic}")
    private String commonOperationTopic;

    /**
     * 获取kafka请求头
     * @return
     */
    private static Map<String, Object> getKafkaHeader() {
        Map<String, Object> headers = new HashMap<>(1);
        headers.put(KafkaHeaderConstant.HEADER_KEY_PLATFORM, KafkaHeaderConstant.HEADER_PLATFORM_VALUE);
        return headers;
    }

    /**
     * 获取发送的 record
     * @param topic
     * @param value
     * @return
     */
    private static ProducerRecord<String, String> getRecord(String topic, String value) {
        GenericMessage<String> message = new GenericMessage<>(value, getKafkaHeader());
        return (ProducerRecord<String, String>) converter.fromMessage(message, topic);
    }

    /**
     *
     * @param cacheOperationVOList 推送刷新缓存数据集合
     */
    public void sendCommonOperationData(List<CacheOperationVO> cacheOperationVOList){
        if(CollectionUtils.isEmpty(cacheOperationVOList)){
            return;
        }
        String value = JacksonUtil.toJSONString(cacheOperationVOList);
        ProducerRecord<String, String> record = getRecord(commonOperationTopic, value);
        template.send(record);
    }

    public List<CacheOperationVO> getSendData(OperationTypeEnum opType, List<Object> data, CacheTopicType cacheTopicType){
        CacheOperationVO cacheOperationVO = new CacheOperationVO();
        cacheOperationVO.setOpType(opType);
        cacheOperationVO.setData(data);
        cacheOperationVO.setTopicType(cacheTopicType);
        cacheOperationVO.setTime(System.currentTimeMillis());
        return Lists.newArrayList(cacheOperationVO);
    }
}
